/*
 *  ------------------------------------------------------------------
 *  Copyright © 2017 Hangzhou DtDream Technology Co.,Lt d. All rights reserved.
 *  ------------------------------------------------------------------
 *              Product: EMR
 *      Module Name: DataTrain
 *      Date Created: 2017-10-23
 *         Description:
 *  ------------------------------------------------------------------
 *  Modification History
 *  DATE            Name           Description
 *  ------------------------------------------------------------------
 *  2017-10-23     XZ 209
 *  ------------------------------------------------------------------
 */

package org.xukai.remoting.sdk.rpc.job;

import com.dtdream.emr.transmission.exception.TransmissionException;

import java.util.*;
import org.xukai.remoting.sdk.rpc.reader.BufferedReader;
import org.xukai.remoting.sdk.rpc.reader.Reader;
import org.xukai.remoting.sdk.rpc.transfer.TransferNode;

import static com.dtdream.emr.transmission.exception.TransmissionException.Error.EXCEPTION_READER_INVALID_ID;


public class DownloadJob extends Job {

    private Map<Integer, Reader> readerMap = new HashMap<>();
    private int readerNum = -1;

    /**
     * 创建下载Job
     *
     * @param properties
     * @throws TransmissionException
     */
    public DownloadJob(Properties properties) throws TransmissionException {
        super(properties);
    }

    protected final UUID createJobInRemote() throws TransmissionException {
        return transferClient.createDownloadJob(getProjectName(), getTableName(), getPartition());
    }

    public synchronized BufferedReader newReader(int id) throws TransmissionException {
        if (id < getReaderNum()) {
            TransferNode node = transferClient.createReader(getJobId(), id);
            BufferedReader bufferedReader = new BufferedReader(this, id, node);
            readerMap.put(id, bufferedReader);
            return bufferedReader;
        } else {
            throw new TransmissionException(EXCEPTION_READER_INVALID_ID, "readerId must be small than " + getReaderNum());
        }
    }

    private int getReaderNum() throws TransmissionException {
        if (readerNum == -1) {
            synchronized (this) {
                if (readerNum == -1) {
                    readerNum = transferClient.getReaderNum(getJobId());
                }
            }
        }

        return readerNum;
    }

    public synchronized Collection<Reader> getReaders() throws TransmissionException {
        for (int i = 0; i < getReaderNum(); i++) {
            newReader(i);
        }

        return readerMap.values();
    }

    public synchronized Reader getReader(int id) {
        return readerMap.get(id);
    }
}
